package net.yuyoo.uusession.core;

import java.io.UnsupportedEncodingException;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.Iterator;
import java.util.Map;
import java.util.SortedMap;
import java.util.TreeMap;

import net.yuyoo.uusession.support.HashUtil;

/**
 * 一致性hash环
 * @author yuyoo (yuyoo4j@163.com)
 * @teme 2010-6-10 下午10:03:29
 * @param <T> -- 节点对象类型
 */
public class ConsistentHashRing<T> {
	 	
	/**
	 * 节点集合
	 */
	protected SortedMap<Long, T> nodes = null;
	
	
	/**
	 * 
	 * @param key
	 * @return
	 */
	public long toConsistentHash(String key) {
		return HashUtil.getConsistentHash(key);
	}		

	public T locate(String key) {
		long h = toConsistentHash(key);
		return locate(h);
	}
	 
	public T locate(long hashCode) {
		
		long h = toConsistentHash(hashCode);
		Long k = Long.valueOf(h);
		SortedMap<Long, T> tails = nodes.tailMap(k);
		if (tails.size() > 0) {
			return nodes.get(tails.firstKey());
		} else { 
			return nodes.get(nodes.firstKey());
		} 
	}
	
	/**
	 * 将hashCode转成一致性hash的值范围[0, 2^32)=[0, 4294967295]
	 * @param hashCode -- hash值
	 * @return -- 一致性hash值
	 */
	public long toConsistentHash(long hashCode) {
		return hashCode  & 0xffffffffL;
	}

	/**
	 * 进行key的值范围检查
	 */
	public void setNodes(Map<Long, T> ns) { 	
					
		TreeMap<Long, T> n = new TreeMap<Long, T>(ns);
		if (!n.isEmpty()) {
			// 检查hash值范围
			Long firstKey = n.firstKey();
			Long lastKey = n.lastKey();
			checkHashRange(firstKey.longValue());
			checkHashRange(lastKey.longValue());			
		}
		this.nodes = n;
	}
	

	
	/**
	 * 增加节点
	 * @param no -- 节点编号
	 * @param node -- 节点
	 */
	public void addNode(long no, T node) {	
		
		if (null == nodes) {
			nodes = new TreeMap<Long, T>();
		}
		checkHashRange(no);
		nodes.put(Long.valueOf(no), node);
	}
	
	/**
	 * 移除节点
	 * @param no -- 节点编号
	 * @return -- 节点, 如果没有节点返回null
	 */
	public Object removeNode(long no) {			
		checkHashRange(no);
		return this.nodes.remove(Long.valueOf(no));
	} 
	
	/**
	 * 获取环顺时针节点迭代器
	 * @param start -- 开始节点编号
	 * @param startInclude -- 迭代器是否包含开始节点
	 * @return -- 迭代器
	 */
	public Iterator<T> clockwiseNodeIterator(long start, boolean startInclude) {

		checkHashRange(start);
		long no = startInclude ? start : start + 1;
		ClockwiseNodeIterator<T> it = new ClockwiseNodeIterator<T>(no, nodes, nodes.size());
		return it;
	}
	
	/**
	 * 获取环顺时针节点迭代器
	 * @param key -- key
	 * @return -- 迭代器
	 */
	public Iterator<T> clockwiseNodeIterator(final String key) {
		
		long k = toConsistentHash(key);
		ClockwiseNodeIterator<T> it = new ClockwiseNodeIterator<T>(k, nodes, nodes.size());
		return it;
	}
	
	/**
	 * 获取环逆时针节点迭代器
	 * @param start -- 开始节点编号
	 * @param startInclude -- 迭代器是否包含开始节点
	 * @return -- 迭代器
	 */
	public Iterator<T> counterclockwiseNodeIterator(long start, boolean startInclude) {

		checkHashRange(start);
		long no = startInclude ? start : start - 1;
		CounterclockwiseNodeIterator<T> it = new CounterclockwiseNodeIterator<T>(no, nodes, nodes.size());
		return it;
	}
	
	/**
	 * 获取环逆时针节点迭代器
	 * @param key -- key
	 * @return -- 迭代器
	 */
	public Iterator<T> counterclockwiseNodeIterator(final String key) {
		
		long k = toConsistentHash(key);
		CounterclockwiseNodeIterator<T> it = new CounterclockwiseNodeIterator<T>(k, nodes, nodes.size());
		return it;
	}
	

	/**
	 * 引用一个节点编号获取下一个获选编号
	 * 备注: 用于分摊一个引用节点的请求压力时使用
	 * @param quote -- 引用节点
	 * @return -- 候选节点编号
	 */
	public long nextCandidate(long quote) {
		
		if (nodes.isEmpty()) {
			throw new IllegalStateException("一致性hash环中没有节点");
		}
		checkHashRange(quote); // 检查引用节点的值范围
		Long quoteKey = Long.valueOf(quote);
		if (!nodes.containsKey(quoteKey)) { // 引用节点不存在
			throw new IllegalStateException("一致性hash环中不存在引用节点(quote):" + quote);
		}
		
		SortedMap<Long, T> heads = nodes.headMap(quoteKey);
		if (heads.isEmpty()) { 
			if (0 == quote) {
				Long last = nodes.lastKey();
				return (4294967296L - last.longValue()) / 2;  // 如果引用节点为0,则转成2^32(4294967296L)值来计算
			} else {
				return quote / 2;
			} 
		} else {
			Long last = heads.lastKey();
			return (quote - last.longValue()) / 2;
		}
	}
	
	/**
	 * 返回指定数量的候选节点编号
	 * @param count -- 指定的数量
	 * @return -- 候选节点编号数组
	 */
	public long[] nextCandidates(int count) {
		
		if (count < 1) {
			throw new IllegalArgumentException("候选数据不能少于1");
		}	
		
		SortedMap<Long, Object> ns = new TreeMap<Long, Object>();
		if (null != nodes) {
			ns.putAll(nodes);
		}
		long[] candidates = new long[count];
		for (int i = 0; i < count; i++) {
			long next = nextCandidate(ns);
			candidates[i] = next;
			ns.put(Long.valueOf(next), new Object());
		}
		return candidates;
	}
	
	/**
	 * 获取下一个候选节点编号
	 * @param ns -- 环节点集合
	 * @return -- 候选节点编号
	 */
	private long nextCandidate(SortedMap<Long, Object> ns) {
		
		if (ns.isEmpty()) { // 环中没有节点
			return 0L;
		} else if (ns.size() == 1) { // 环中节点数为 1
			long first = ns.firstKey().longValue();
			if (first > 2147483648L) {  // 超过 2^32(4294967296L)值的一半了,说明下个点在对面
				return first - 2147483648L;
			} else {
				return first + 2147483648L;
			}
		}
        // 环中节点数大于1
		
		// 算出环中每两个节点的距离
		long last = ns.firstKey().longValue();
		SortedMap<Long, Object> tails = ns.tailMap(Long.valueOf(last + 1));
		long [] ranges = new long[ns.size()]; 
		int index = 0;
		for (Long key : tails.keySet()) {
			long next = key.longValue();
			long range = next - last; 
			last = next;
			ranges[index] = range;
			index++;
		}
		// 算最后一个节点与第一个节点的距离
		ranges[ranges.length - 1] = 4294967296L - ns.lastKey().longValue() + ns.firstKey().longValue();
		
		// 求对大的距离与在集合中的偏移值
		int maxIndex = 0;
		long maxRange = 0;
		for (int i = 0; i < ranges.length; i++) {
			if (ranges[i] > maxRange) {
				maxRange = ranges[i];
				maxIndex = i;
			}
		}
		// 定位最大距离的节点值
		Iterator<Long> it = ns.keySet().iterator();
		Long nodeIndex = null;
		for (int i = 0; i <= maxIndex; i++) {
			nodeIndex = it.next();
		}
		
		// 算最大距离的候选节点值
		long c = nodeIndex.longValue() + maxRange / 2;
		if (c < 4294967296L) { // 候选节点值没有超出一致性hash环值的范围
			return c;
		} else if (c > 4294967296L) {
			return ns.firstKey() - maxRange / 2;
		} else {
			return 0L;
		} 
	}
	
	/**
	 * 检查hash值是否满足一致性hash的值范围要求
	 * @param key -- 被检查的hash值
	 */
	private void checkHashRange(long hash) {
		if (hash < 0) {
			throw new IllegalArgumentException("一致性hash的hash值范围[0, 2^32)=[0, 4294967295], 不能存在小于0的节点:" + hash);
		} 
		if (hash > 4294967295L) {
			throw new IllegalArgumentException("一致性hash的hash值范围[0, 2^32)=[0, 4294967295], 不能存在大于4294967295的节点:" + hash);
		}
	}

	/**
	 * 顺时针节点迭代器
	 * @author zhandl(zhandl@hainan.net)
	 * @time 2009-12-21 上午10:17:19
	 */
	public static class ClockwiseNodeIterator<T> implements Iterator<T> {
		
		private int remaining = 0;
		
		private SortedMap<Long, T> nodes = null;

		private long currentKey = -1;
		
		public ClockwiseNodeIterator(long current, SortedMap<Long, T> ns, int remaining) {
			
			this.remaining = remaining;
			this.nodes = ns;
			this.currentKey = current;
		}

		public boolean hasNext() {
			return (remaining > 0);
		}

		public T next() {
			
			remaining--;
			T n = nodes.get(currentKey);
			long nextKey = currentKey + 1;
			if (null == n) {
				SortedMap<Long, T> tails = nodes.tailMap(currentKey);
				Long firstKey = null;
				if (tails.isEmpty()) {
					firstKey = nodes.firstKey();
					n = nodes.get(firstKey); 
				} else {
					firstKey = tails.firstKey();
					n = nodes.get(firstKey);
				}
				nextKey = firstKey.intValue() + 1;
			} 
			currentKey = nextKey;
			return n;
		}

		public void remove() {				
			throw new UnsupportedOperationException("不支持此操作");
		} 
					
		@Override
		public String toString() {

			StringBuilder sb = new StringBuilder();
			sb.append("ClockwiseNodeIterator{"); 
			sb.append("currentKey").append(currentKey).append(",");
			sb.append("remaining").append(remaining).append(",");
			sb.append("nodes").append(nodes);
			sb.append("}");
			return sb.toString();
		}
	}
	
	/**
	 * 逆时针节点迭代器
	 * @author zhandl(zhandl@hainan.net)
	 * @time 2009-12-21 上午10:20:28
	 */
	public static class CounterclockwiseNodeIterator<T> implements Iterator<T> {
		
		private SortedMap<Long, T> nodes = null; 
		private long currentKey = -1;
		private int remaining = 0;
		
		public CounterclockwiseNodeIterator(long current, SortedMap<Long, T> nodes, int remaining) {
			  
			this.currentKey = current;
			this.nodes = nodes;
			this.remaining = remaining;
		}

		public boolean hasNext() {
			
			return remaining > 0;
		}

		public T next() {
			
			remaining--;
			T n = nodes.get(currentKey);
			long nextKey = currentKey - 1;
			if (null == n) {
				SortedMap<Long, T> heads = nodes.headMap(Long.valueOf(currentKey + 1));
				Long lastKey = null;
				if (heads.isEmpty()) {
					lastKey = nodes.lastKey();
					n = nodes.get(lastKey); 
				} else {
					lastKey = heads.lastKey();
					n = nodes.get(lastKey);
				}
				nextKey = lastKey.intValue() - 1;
			} 
			currentKey = nextKey;
			return n;
		}

		public void remove() {
			throw new UnsupportedOperationException("不支持移除操作");
		}	 	
		
		@Override
		public String toString() {

			StringBuilder sb = new StringBuilder();
			sb.append("CounterclockwiseNodeIterator["); 
			sb.append("currentKey").append(currentKey).append(",");
			sb.append("remaining").append(remaining).append(",");
			sb.append("nodes").append(nodes);
			sb.append("]");
			return sb.toString();
		}
	}
}
